package org.replay.service.impl;

/*
 * Copyright © 2020 lizicong,All rights reserved.
 *
 * Licensed under the GPL License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.gnu.org/licenses/old-licenses/gpl-2.0.html
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import io.lettuce.core.api.sync.RedisCommands;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.CreateMode;
import org.replay.config.RabbitBindConfiguration;
import org.replay.repository.RedisDynamicCommands;
import org.replay.service.SeckillCommodityService;
import org.replay.utils.OrderNumGeneration;
import org.replay.utils.RabbitACKSender;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Date;

/**
 * 商品秒杀模块服务接口实现
 *
 * @author lizicong
 */
@Service
public class SeckillCommodityServiceImpl implements SeckillCommodityService {

    /**
     * 是否开启强制校验只能秒杀一次功能?
     */
    final static boolean is_JUST_ONLY_BUY_ONE = true;

    final static String ZK_PREFIX = "/SECKILL/STOCK/COMMODITY";
    final static String ZK_LOCK_NAME = "/LOCKSET";
    final static String ZK_LISTENER_EVENT = "CHILD_REMOVED";
    final static String REDIS_PREFIX = "SECKILL_STOCK_COMMODITY_";
    final static String REDIS_SUFFIX = "_BOUGHT_USER_ID";

    /**
     * lettuce动态命令客户端
     * 未限制读操作
     */
    @Resource
    RedisDynamicCommands redisDynamicCommands;

    /**
     * lettuce常规命令客户端
     * 已限制读操作
     */
    @Resource
    RedisCommands redisCommands;

    /**
     * MQ手动ACK签收器
     */
    @Resource
    RabbitACKSender rabbitACKSender;

    /**
     * zk客户端
     */
    @Resource
    CuratorFramework curatorFramework;

    @Override
    public String placeAnOrder(Integer userID, Integer commodityID) {
        String commodityKeyName = REDIS_PREFIX + commodityID;
        String commodityBoughtUserIDKeyName = commodityKeyName + REDIS_SUFFIX;
        //检查用户是否已购买过？(具有幂等性的插入)
        if (redisDynamicCommands.sadd(commodityBoughtUserIDKeyName, userID) == 1) {
            //检查库存是否足够？
            if (Integer.valueOf(redisCommands.get(commodityKeyName).toString()) > 0) {
                //检查是否得到锁？
                if (isGetLock(userID, commodityID, this)) {
                    //检查是否修改库存成功？
                    if (redisDynamicCommands.set(commodityKeyName, Integer.valueOf(redisDynamicCommands.get(commodityKeyName)) - 1).equals("OK")) {
                        String msg = userID + "/" + commodityID + "/" + new OrderNumGeneration().getNum().toString() + "/" + new Date().toString();
                        rabbitACKSender.rabbitTemplate.convertAndSend(RabbitBindConfiguration.getMyExchangeName(), RabbitBindConfiguration.getMyRouteKey(), msg, new CorrelationData(msg));
                        delLock(commodityID.toString());
                        return "ALL OK";
                    } else {
                        System.out.println("Failed to buy");
                    }
                }
            } else {
                System.out.println("Inventory is empty");
            }
        } else {
            System.out.println("You can't buy goods repeatedly");
        }
        return "NOT OK";
    }


    @Override
    public void nodePreheat() {
        isGetLock(0, 0, null);
        delLock("0");
        String msg = 0 + "/" + 0 + "/" + new OrderNumGeneration().getNum().toString() + "/" + new Date().toString();
        rabbitACKSender.rabbitTemplate.convertAndSend(RabbitBindConfiguration.getMyExchangeName(), RabbitBindConfiguration.getMyRouteKey(), msg, new CorrelationData(msg));
        for (int i = 1; i <= 4; i++) {
            redisDynamicCommands.set(REDIS_PREFIX + i, 500);
            redisDynamicCommands.del(REDIS_PREFIX + i + REDIS_SUFFIX);
        }

    }

    /**
     * zk回调监听器
     * 实现分布式乐观锁
     *
     * @param userID                  userID
     * @param commodityID             commodityID
     * @param seckillCommodityService seckillCommodityService
     * @return 是否得到锁？
     */
    private boolean isGetLock(int userID, int commodityID, SeckillCommodityServiceImpl seckillCommodityService) {
        //监听路径
        String listenerPath = ZK_PREFIX + "/" + commodityID;
        //锁节点路径
        String lockNodePath = listenerPath + ZK_LOCK_NAME;
        //尝试注册节点
        try {
            curatorFramework.create().withMode(CreateMode.EPHEMERAL)
                    .forPath(lockNodePath);
        }
        //节点存在,注册该节点监听器
        catch (Exception e) {
            System.out.println("Node already exists,Ready to register ZK listener");
            //手动释放监听器有点麻烦，不过可以交给try()块来直接kill用过的监听器
            try (PathChildrenCache cache = new PathChildrenCache(curatorFramework, listenerPath, true)) {
                cache.start();
                cache.rebuild();
                cache.getListenable().addListener(new PathChildrenCacheListener() {
                    @Override
                    public void childEvent(CuratorFramework framwork, PathChildrenCacheEvent event) {
                        System.err.println("***Listener callback,Event: " + event.getType());
                        if (event.getType().toString().equals(ZK_LISTENER_EVENT))
                            seckillCommodityService.placeAnOrder(userID, commodityID);
                    }
                });
                return false;
            } catch (Exception e2) {
                e.printStackTrace();
                e2.printStackTrace();
                return false;
            }
        }
        return true;
    }

    /**
     * 删除锁节点
     *
     * @param commodityID 商品ID
     */
    private void delLock(String commodityID) {
        try {
            curatorFramework.delete().forPath(ZK_PREFIX + "/" + commodityID + ZK_LOCK_NAME);
        } catch (Exception e) {
        }
    }
}
